package com.ikas.autumn.iec104.common.config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * @author hxh
 * @ClassName: ProducerKafkaConfig
 * @Description: 生产者配置
 * @date 2022年03月01日
 */
@Configuration
@ConditionalOnProperty(prefix = "enable", name = "kafka", havingValue = "true", matchIfMissing = false)
public class KafkaProducerConfig {

    /**
     * 创建Kafka服务器的连接
     */
    @Value("${spring.kafka.producer.bootstrap-servers}")
    private String bootstrapServers;
    /**
     * 消息发送失败最大尝试次数
     */
    @Value("${spring.kafka.producer.retries:1}")
    private String retries;
    /**
     * 批量发送消息的数据包大小
     */
    @Value("${spring.kafka.producer.batch-size:524288}")
    private String batchSize;
    /**
     * 消息延时发送时间
     */
    @Value("${spring.kafka.producer.linger-ms:10}")
    private String lingerms;
    /**
     * 缓存消息的缓冲区大小
     */
    @Value("${spring.kafka.producer.buffer-memory:536870912}")
    private String bufferMemory;

    @Value("${spring.kafka.producer.max-block-ms:5000}")
    private String maxBlockMsConfig;
    /**
     * 压缩消息的格式
     */
    @Value("${spring.kafka.producer.compression-type:lz4}")
    private String compressionType;

    @Value("${spring.kafka.producer.request-timeout-ms:120000}")
    private String requestTimeoutmsConfig;

    @Value("${spring.kafka.producer.max-request-size:12582912}")
    private String maxRequestSizeConfig;
    /**
     * 消息key类型
     */
    private String keySerializerClass = "org.apache.kafka.common.serialization.StringSerializer";
    /**
     * 消息value类型
     */
    private String valueSerializerClass = "org.apache.kafka.common.serialization.StringSerializer";

    /**
     * 控制生产者的消息持外性
     */
    @Value("${spring.kafka.producer.acks:1}")
    private String acks;

    /**
     * @param @return 参数
     * @return Map<String, Object> 返回类型
     * @throws
     * @Title: producerConfigs
     * @Description: 初始化生产者配置参数
     */
    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.ACKS_CONFIG, acks);
        props.put(ProducerConfig.RETRIES_CONFIG, retries);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
        props.put(ProducerConfig.LINGER_MS_CONFIG, lingerms);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType);
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, ComstomPartitioner.class.getName());
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializerClass);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializerClass);
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, maxBlockMsConfig);
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutmsConfig);
        return props;
    }

    /**
     * Producer 工厂配置
     */
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    /**
     * Producer Template 配置
     */
    @Bean(name="kafkaTemplate")
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}
